package cn.lingyangwl.agile.msg.push.core.propcess;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.func.LambdaUtil;
import cn.lingyangwl.agile.dynamic.config.core.*;
import cn.lingyangwl.agile.infra.*;
import cn.lingyangwl.agile.infra.module.group.model.rqrs.*;
import cn.lingyangwl.agile.model.enums.*;
import cn.lingyangwl.agile.model.module.config.*;
import cn.lingyangwl.agile.model.module.config.param.*;
import cn.lingyangwl.agile.msg.push.api.constatnts.*;
import cn.lingyangwl.agile.msg.push.api.model.content.*;
import cn.lingyangwl.agile.msg.push.api.model.rqrs.*;
import cn.lingyangwl.agile.msg.push.core.*;
import cn.lingyangwl.agile.msg.push.core.sender.*;
import cn.lingyangwl.agile.msg.push.manager.*;
import cn.lingyangwl.agile.msg.push.model.*;
import cn.lingyangwl.agile.msg.push.model.entity.*;
import cn.lingyangwl.agile.msg.push.service.*;
import cn.lingyangwl.framework.cache.core.service.*;
import cn.lingyangwl.framework.tool.core.*;
import cn.lingyangwl.framework.tool.core.exception.*;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.util.UriComponentsBuilder;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author shenguangyang
 */
@Slf4j
@Service
public class MsgPushProcessImpl implements MsgPushProcess {
    @Resource
    private MsgHistoryManager historyManager;
    @Resource
    private MsgSendManager msgSendManager;
    @Resource
    private CacheService cacheService;
    @Resource
    private NoticeChannelService noticeChannelService;
    @Resource
    private InfraApi infraApi;
    @Resource
    private MsgNoticeService noticeService;
    @Resource
    private ReceiverSubscribeManager receiverSubscribeManager;
    @Resource
    private ConfigFactory configFactory;

    @Override
    public void push(MsgPushParam req, String msgId) throws RuntimeException {
        PushTaskContext context = new PushTaskContext();
        // 解析消息内容
        MsgParam msgParam = req.getMsgParam();
        context.setMsgParam(msgParam);
        Assert.notEmpty(req.getChannelType(), "渠道类型不能为空");

        // 通过编码获取消息通知
        MsgNotice msgNotice = noticeService.getNoticeByCode(msgParam.getCode());
        if (Objects.isNull(msgNotice)) {
            throw new BizException("指定的通知 [{}] 不存在", msgParam.getCode());
        }

        Set<String> toUserList = new HashSet<>();

        // 处理当前任务的接收者类型为通知管理员
        if (Objects.equals(msgNotice.getReceiverType(), MsgCons.ReceiverType.ADMIN)) {
            List<Receiver> receiverList = receiverSubscribeManager.listMsgSubscriber(msgNotice.getId());
            Set<String> toAdminUserList = receiverList.stream().flatMap(e -> ReceiverConfig.of(e).stream())
                    .filter(e -> Objects.equals(e.getChannelType().getCode(), req.getChannelType()))
                    .map(ReceiverConfig::getToUser).collect(Collectors.toSet());
            if (StringUtils.isEmpty(toAdminUserList)) {
                log.warn("推送的任务 [{}] 是管理员通知类型, 但是没有任何接收者验证接受渠道信息", msgNotice.getCode());
                return;
            }
            toUserList.addAll(toAdminUserList);
        } else {
            if (StringUtils.isEmpty(msgParam.getToUser())) {
                throw new BizException("当前任务非管理员推送类型, 需要调用方指定接收者");
            }
            toUserList.add(msgParam.getToUser());
        }

        // 获取任务关联的模板并发送消息
        List<NoticeChannel> channelList = noticeChannelService.listChannel(msgNotice.getId(), req.getChannelType());
        log.info("noticeCode: {}, channelType: {}, channelSize: {}", msgParam.getCode(), req.getChannelType(), channelList.size());
        if (CollUtil.isEmpty(channelList)) {
            String channelTypeName = Optional.ofNullable(ChannelTypeEnum.ofByCode(req.getChannelType()))
                .map(ChannelTypeEnum::name).orElse("*");
            log.error("尚未启用渠道类型 [{}]", channelTypeName);
            return;
        }
        for (NoticeChannel noticeChannel : channelList) {
            MsgHistory history = new MsgHistory();
            String groupName = null;
            MsgSettings msgSettings = JSON.parseObject(noticeChannel.getSettings(), MsgSettings.class);
            // 获取标题
            msgParam.setTitle(StringUtils.priorityFirstNotEmpty(msgParam.getTitle(), msgSettings.getTitle(), msgNotice.getName()));
            ChannelTypeEnum channelTypeEnum = ChannelTypeEnum.ofByCode(noticeChannel.getChannelType());
            for (String toUser : toUserList) {
                try {
                    history.setChannelId(noticeChannel.getId()).setPushTime(new Date())
                            .setGroupId(msgNotice.getGroupId()).setToUser(toUser);
                    context.getMsgParam().setToUser(toUser);
                    if (Objects.isNull(channelTypeEnum)) {
                        log.warn("未查找到 channelType: {} 的渠道编码", noticeChannel.getChannelType());
                        throw new BizException("channelType [{}] 不存在", noticeChannel.getChannelType());
                    }
                    groupName = Optional.ofNullable(infraApi.getDataGroup(msgNotice.getGroupId())).map(DataGroupBase::getName)
                            .orElse("未知");

                    context.setChannelType(channelTypeEnum);

                    // 推送消息
                    MsgSender sendHandler = msgSendManager.getMsgSendHandler(channelTypeEnum);

                    // 初始化消息设置
                    msgSettings.parseField(context.getMsgParam().getMsgFieldList());
                    msgSettings.evaluate();

                    // 默认采用传进来的url, 否则采用后台配置的

                    msgParam.setPageUrl(getPageUrl(msgSettings, msgParam));

                    context.setMsgSettings(msgSettings);
                    context.setPushStatus(MsgPushStatus.SUCCESS);
                    sendHandler.sendMessage(context);
                    if (Objects.isNull(context.getChannelCode())) {
                        throw new BizException("channelCode is null");
                    }

                    // 因为当前方法消息队列监听器也会调用, 所以要抛出异常, 触发重试
                    if (MsgPushStatus.FAIL.equals(context.getPushStatus())) {
                        throw new BizException("推送失败 [{}]", context.getPushResult());
                    }
                } catch (Exception e) {
                    log.error("error: ", e);
                    context.setPushResult(StringUtils.defaultIfEmpty(context.getPushResult(), e.getMessage()));
                    context.setPushStatus(MsgPushStatus.FAIL);
                    throw e;
                } finally {
                    try {
                        // 过滤使用的字段
                        List<MsgField> msgFields = msgParam.getMsgFieldList().stream()
                            .filter(e -> msgSettings.getTplField().containsKey(e.getKey()))
                            .collect(Collectors.toList());
                        String channelCode = Optional.ofNullable(context.getChannelCode()).map(ChannelCodeEnum::getCode).orElse("-1");
                        String channelType = Optional.ofNullable(context.getChannelType()).map(ChannelTypeEnum::getCode).orElse("-1");
                        history.setPushResult(context.getPushResult()).setId(req.getHistoryId()).setNoticeId(msgNotice.getId())
                            .setBizKey(msgParam.getBizKey()).setBizType(msgParam.getBizType()).setTitle(msgParam.getTitle())
                            .setStatus(context.getPushStatus().getStatus()).setGroupName(groupName)
                            .setChannelCode(channelCode).setChannelType(channelType)
                            .setPageUrl(context.getMsgParam().getPageUrl()).setMsgFields(msgFields);
                        save(history, msgId);
                    } catch (Exception e) {
                        if (e instanceof NullPointerException) {
                            log.error("error: ", e);
                        }
                    } finally {
                        if (context.getPushStatus() == MsgPushStatus.FAIL) {
                            log.error("消息推送失败, content: {}", JSON.toJSONString(context));
                        }
                    }
                }
            }

        }
    }

    /**
     * 优先使用调用方传递的页面地址, 否则采用后台配置的
     */
    public String getPageUrl(MsgSettings msgSettings, MsgParam content) {
        String pageUrl = StringUtils.defaultIfEmpty(content.getPageUrl(), msgSettings.getPageUrl()).replaceAll(" ", "");
        if (pageUrl.startsWith("${")) {
            ConfigInfo configInfo = configFactory.getConfig(ConfigDefineEnum.WEB_SITE);
            if (Objects.nonNull(configInfo)) {
                WebSiteParam param = BaseConfigParam.jsonToObject(configInfo.getConfigKey(), configInfo.getConfigValue());
                String var = "${" + LambdaUtil.getFieldName(WebSiteParam::getH5Url) + "}";
                pageUrl = pageUrl.replace(var, param.getH5Url());
            }
        }
        if (StringUtils.isEmpty(pageUrl)) {
            return null;
        }
        String agreement = "";
        if (pageUrl.startsWith("http://")) {
            agreement = "http://";
        } else if (pageUrl.startsWith("https://")) {
            agreement = "https://";
        }
        pageUrl = pageUrl.replace(agreement, "");
        Map<String, String> urlParams = content.getUrlParams();
        UriComponentsBuilder uriBuilder = UriComponentsBuilder.fromPath(pageUrl);

        MultiValueMap<String, String> queryParams = new LinkedMultiValueMap<>();
        urlParams.forEach((k, v) -> queryParams.put(k, Collections.singletonList(v)));

        uriBuilder.queryParams(queryParams);

        return agreement + uriBuilder.build();
    }

    /**
     * 防止消费失败, 多次重试往数据库中插入重复数据, 如果多次重试之后, 依旧失败只保留最后一次的失败信息, 如果经过
     * 几次重试之后, 消费成功, 则保留成功的信息
     */
    private void save(MsgHistory history, String msgId) {
        if (Objects.nonNull(history.getId())) {
            MsgHistory entity = historyManager.getById(history.getId());
            history.setRetryCount(entity.getRetryCount() + 1);
            historyManager.updateById(history);
            return;
        }

        String key = MsgPushCacheKey.MSG_PUSH_SAVE.formatKey(msgId);
        Object value = cacheService.opsForValue().get(key);
        if (Objects.nonNull(value)) {
            history.setId(Long.parseLong(String.valueOf(value)));
            MsgHistory po = historyManager.getById(history.getId());
            history.setRetryCount(po.getRetryCount() + 1);
            historyManager.updateById(history);
            return;
        }
        historyManager.save(history);
        cacheService.opsForValue().set(key, history.getId());
    }
}
